package com.xinqing.bigdata.flink.datastream.operators;

import com.xinqing.bigdata.flink.datastream.model.Worker;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.Properties;

/**
 * @Author:CHQ
 * @Date:2021/5/20 16:52
 * @Description
 */
public class Union {
    public static void main(String args[]) throws Exception {
        //env,port
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());

        env.setParallelism(1);

        //checkpoint conf
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        checkpointConfig.setCheckpointTimeout(60000);
        checkpointConfig.setCheckpointInterval(60000);
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        Tuple2<String, String> tuple2 = new Tuple2<>();
        tuple2.setFields("1", "2");
        DataStreamSource<Tuple2<String, String>> tuple2DataStreamSource2 = env.fromElements(tuple2);


        SingleOutputStreamOperator<Tuple2<String, String>> map = tuple2DataStreamSource2.map(new MapFunction<Tuple2<String, String>, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(Tuple2<String, String> value) throws Exception {
                return new Tuple2<>(Integer.toString(Integer.parseInt(value.f0) + 1), Integer.toString(Integer.parseInt(value.f1) + 1));
            }
        });

        tuple2DataStreamSource2.union(map).print();

        env.execute();

    }
}
